package com.tally.user.mq;

import com.alibaba.fastjson.JSON;
import com.tally.server.dto.SysExpDetailDTO;
import com.tally.server.service.ExpDetailService;
import com.tally.server.service.ExpService;
import com.tally.server.service.SysExpDetailService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;

/**
 * @Author: sy
 * @Date: 2021/2/3 22:48
 * @Description:
 */
@Component
public class MqConsumer {
    private Logger logger = LoggerFactory.getLogger(MqConsumer.class);

    private DefaultMQPushConsumer consumer;
    @Value(value = "${mq.nameserver.addr}")
    private String nameAddr;
    @Value(value = "${mq.topicname}")
    private String topicName;

    @Autowired
    private SysExpDetailService sysExpDetailService;

    @Autowired
    private ExpDetailService expDetailService;

    @Autowired
    private ExpService expService;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer("mq_consumer_group");
        consumer.setNamesrvAddr(nameAddr);
        consumer.subscribe(topicName,"*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            @Transactional
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                Message message = list.get(0);
                String jsonString = new String(message.getBody());
                Map<String,Object> map = JSON.parseObject(jsonString,Map.class);
                Integer userId = (Integer) map.get("userId");
                String type = (String) map.get("type");
                SysExpDetailDTO sysExpDetailDTO = sysExpDetailService.getByType(type);
                Integer expValue = expDetailService.addExp(userId, sysExpDetailDTO);
                expService.addExp(userId,expValue);
                logger.info("加经验完成，userId={},exp={}",userId,expValue);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

}
